package cn.edu.recommender

import org.apache.spark.sql.SparkSession

import java.lang.Thread.sleep
import java.sql.Date
import java.text.SimpleDateFormat


case class Rating(userId: Int, productId: Int, score: Double, timestamp: Int)

case class MongoConfig(uri: String, db: String)


object StatisticsRecommender {

  val MONGODB_RATING_COLLECTION = "Rating"
  //统计的表的名称
  val RATE_MORE_PRODUCTS = "RateMoreProducts"
  val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts"
  val AVERAGE_PRODUCTS = "AverageProducts"

  def main(args: Array[String]): Unit = {
    // 定义用到的配置参数
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://localhost:27017/recommender",
      "mongo.db" -> "recommender"
    )

    val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 创建一个 SparkSession
    val spark = SparkSession.builder().appName("StatisticsRecommender").master(config("spark.cores")).getOrCreate()

    // 在对 DataFrame 和 Dataset 进行操作许多操作都需要这个包进行支持
    import spark.implicits._

    //数据加载进来
    val ratingDS = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating]


    //创建一张名叫 ratings 的表
    ratingDS.createOrReplaceTempView("ratings")
//    spark.catalog.cacheTable("ratings")

    //统计所有历史数据中每个商品的评分数
    //数据结构 -> productId,count
    val rateMoreProductsDF = spark.sql(
      s"""
         |select productId, count(productId) as count
         |from ratings
         |group by productId
         |""".stripMargin)

    rateMoreProductsDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", RATE_MORE_PRODUCTS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()


    //统计以月为单位拟每个商品的评分数
    //数据结构 -> productId,count,time

    //创建一个日期格式化工具
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")
    //注册一个 UDF 函数，用于将 timestamp 装换成年月格式 1260759144000 => 201605
    spark.udf.register("changeDate", (x: Int) => simpleDateFormat.format(new Date(x * 1000L)).toInt)


    val rateMoreRecentlyProductsDF = spark.sql(
      s"""
         |select productId, count(1) as count, yearmonth
         |from
         |(
         |    select productId, score, changeDate(timestamp) as yearmonth
         |    from ratings
         |) a
         |group by yearmonth, productId
         |order by yearmonth desc, count desc
         |""".stripMargin)

    rateMoreRecentlyProductsDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", RATE_MORE_RECENTLY_PRODUCTS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()


    //统计每个商品的平均评分
    //数据结构 -> productId,avg
    val averageProductsDF = spark.sql(
      s"""
         |select productId, avg(score) as avg
         |from ratings
         |group by productId
         |""".stripMargin)

    averageProductsDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", AVERAGE_PRODUCTS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()


    sleep(300000)
    // 关闭 Spark
    spark.stop()

  }
}
